package com.geniusk.mqtt;

import com.geniusk.mqtt.strategy.MessageStrategy;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * mqtt broker channel 处理
 * ChannelInboundHandlerAdapter:实现了ChannelInboundHandler 的接口
 */
@Component
@Slf4j
@ChannelHandler.Sharable
public class MqttChannelHandler extends ChannelInboundHandlerAdapter  {

    @Autowired
    MessageStrategyManager messageStrategyManager;

    @Override
    public void channelRead(ChannelHandlerContext channelHandlerContext, Object msg) {
        try{
            MqttMessage mqttMessage = (MqttMessage) msg;
            MqttMessageType type = mqttMessage.fixedHeader().messageType();
            MessageStrategy messageStrategy =  messageStrategyManager.getMessageStrategy(type);
            if(messageStrategy!=null){
                messageStrategy.sendResponseMessage(channelHandlerContext.channel(),mqttMessage);
            }
        }catch (Exception e){
          e.printStackTrace();
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 当出现异常就关闭连接
        log.error("Error in class:{}, message:{}", getClass().getSimpleName(), cause.getMessage());
        cause.printStackTrace();
        ctx.close();
    }
}
